package example;

import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.rmi.registry.LocateRegistry;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import javax.management.InstanceAlreadyExistsException;
import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.ObjectName;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;

public class SimpleClient implements SimpleClientMBean{
	static File[] files;
	static AtomicInteger integer=new AtomicInteger(0);
	 int rate=0;
	SlaveClientThread[] threads;
	static String keyspace="";
	static JMXConnectorServer cserver;
	static int simpleK = 1;
	/**
	 * @param args
	 * @throws ClassNotFoundException 
	 * @throws SQLException 
	 * @throws IOException 
	 * @throws NullPointerException 
	 * @throws MalformedObjectNameException 
	 * @throws NotCompliantMBeanException 
	 * @throws MBeanRegistrationException 
	 * @throws InstanceAlreadyExistsException 
	 */
	public static void main(String[] args) throws IOException  {
		
		SimpleClient client=new SimpleClient();
		LocateRegistry.createRegistry(9999);
		MBeanServer server = ManagementFactory.getPlatformMBeanServer();
		String user="root";
		String pw="root";
		String[] credentials=new String[]{user,pw};
		Map<String,String[]> props=new HashMap<String,String[]>();
		props.put("jmx.remote.credentials", credentials);
		JMXServiceURL addr=new JMXServiceURL("service:jmx:rmi:///jndi/rmi://0.0.0.0:9999/jmxrmi");
		cserver =JMXConnectorServerFactory.newJMXConnectorServer(addr, props, server);   
		try{
			cserver.start(); 
			ObjectName worldName=new ObjectName("hwc:type=SimpleClient");
			server.registerMBean(client, worldName);
			client.doMain(args);
		}catch(Exception e){
			e.printStackTrace();
			cserver.stop();
		}
	}
	public void doMain(String args[])throws ClassNotFoundException, SQLException {
		Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
		if(args.length<7){
			System.out.println("use java -jar ***.jar [ip] [port] [mode:0 recreate,1 insert directly] [thread number] [rate number] [keyspace number] [folder] \n");
			return;
		}
		String[] ips=args[0].split(",");
		String port=args[1];
		String mode=args[2];
		Integer threadNumber=Integer.valueOf(args[3]);
		rate=Integer.valueOf(args[4]);
		keyspace=args[5];
		String folder=args[6];
		files=new File(folder).listFiles();
		if(threadNumber>files.length){
			System.out.println("threadNumber is bigger than files ,will set thread number to files number ");
			threadNumber=files.length;
		}
		String url="jdbc:cassandra://"+ips[0]+":"+port;

		if(mode.equals("0")){
			Connection con =null;
			try{
				String createKs="CREATE KEYSPACE test_metrics"+keyspace+" WITH replication " +  "= {'class':'SimpleStrategy', 'replication_factor':3};";
				String createCf="CREATE TABLE test_metrics"+keyspace+".query (" +
						"uid text," + 
						"stime text," + 
						"word text," + 
						"rank int," + 
						"squenceid int," +
						"url text,"+
						"primary key(uid,stime)"+
						");";
				System.out.println("connect:"+url);
				con = DriverManager.getConnection(url);
				Statement statement =con.createStatement();
				try{
					statement.execute("drop keyspace test_metrics"+keyspace);
				}catch(Exception e){

				}
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
				}
				statement.execute(createKs);
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
				}
				statement.execute(createCf);
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
				}
			}catch(Exception e){
				e.printStackTrace();
			}finally{
				con.close();
			}
		}

		 threads=new SlaveClientThread[threadNumber];
		for(int i=0;i<threadNumber;i++){
			threads[i]=new SlaveClientThread(ips,port,this);
		}
		for(int i=0;i<threadNumber;i++){
			threads[i].start();
		}
		//		String query ="SELECT * from sogou.query;";
		//		PreparedStatement statement =con.prepareStatement(query);
		//		ResultSet rs=statement.executeQuery();
		//		while (rs.next()) {
		//			ResultSetMetaData metaData=rs.getMetaData();
		//			int number=metaData.getColumnCount();
		//			for(int i=1;i<=number;i++){
		//				System.out.print(metaData.getColumnName(i)+":"+rs.getObject(i)+"\t");
		//			}
		//			System.out.println();
		//		}

		//	    String query = "UPDATE Test SET a=?, b=? WHERE KEY=?";
		//	    PreparedStatement statement = con.prepareStatement(query);
		//
		//	    statement.setLong(1, 100);
		//	    statement.setLong(2, 1000);
		//	    statement.setString(3, "key0");
		//
		//	    statement.executeUpdate();
		//
		//	    statement.close();

	}
	public static synchronized File requestFile(){
		if(integer.get()>=files.length){
			return null;
		}
		return files[integer.getAndIncrement()];
	}
	@Override
	public void setRate(int rate) {
		this.rate=rate;
	}
	public int getRate(){
		return rate;
	}
	@Override
	public double[] getDynamicRate() {
		// TODO Auto-generated method stub
		double[] ds=new double[threads.length];
		for(int i=0;i<threads.length;i++){
			ds[i]=threads[i].getDynamicRate();
		}
		return ds;
	}
	public void closeServer(){
		try {
			cserver.stop();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	@Override
	public void setSimpleK(int k) {
		simpleK = k;
		
	}
	
	public int getSimpleK() {
		return simpleK;
	}

}
